[Amazon BedrockとSlackで作る生成AIチャットボット] 画像ファイルを生成してくれるチャットボットを作る

みなさん、こんにちは！
福岡オフィスの青柳です。

以前2回に渡って公開したブログ記事「Amazon BedrockとSlackで生成AIチャットボットアプリを作る」を応用して、「画像ファイルを生成してくれるチャットボット」を作りたいと思います。

Slackでチャットボットを作るのは初めてという方は、まずこちらの記事からどうぞ。

今回作成するチャットボット

チャットボットをメンションして、描いて欲しい画像について入力します。

これが浮世絵風？ なのかどうかは少し審議が必要ですが、画像を生成してくれました。

チャットボットを作る

今回作成するチャットボットは、次のようなアーキテクチャです。

以前のブログと全体の構成は同じですが、画像を取り扱うために「バックエンドLambda」の処理内容を少し変更します。

ベースとなるSlackチャットボットを作る

まず、以前のブログを参考にして、シンプルな「生成AIチャットボット」を作成します。

Slackの設定変更

ここから、このチャットボットを「画像ファイルを生成してくれるチャットボット」に進化させるための設定を行なっていきます。

「シンプルな生成AIチャットボット」ではSlackアプリに「応答のテキストを書き込む」ための権限としてchat:writeのスコープを設定していました。

今回は、これに加えて「Slackアプリが画像ファイルをチャンネルに書き込む」ための権限が必要になります。

Slack Appの設定画面で「OAuth & Permissions」を開き、「Bot Token Scopes」にfiles:writeのスコープを追加します。

スコープを追加した後、Slack Appの再インストールを忘れないようにしてください。

Lambda関数の更新 (Slack向けHTTP APIサーバー)

Lambda関数を修正します。 まずは、API Gatewayと統合する「Slack向けHTTP APIサーバー」のLambda関数です。

コード内容は「シンプルな生成AIチャットボット」と基本的に同じです。

Slackのトークン・シークレットを環境変数から取得する箇所を少し変えています。 また、「SQSのキューURL」についても環境変数から取得するように変更しました。 (環境変数SQS_QUEUE_URLを設定してください)

from slack_bolt import App
from slack_bolt.adapter.aws_lambda import SlackRequestHandler
import boto3
import json
import os
import re


sqs_queue_url = os.environ.get("SQS_QUEUE_URL")
slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")
slack_signing_secret = os.environ.get("SLACK_SIGNING_SECRET")

sqs = boto3.client("sqs")


# アプリの初期化
app = App(
    token=slack_bot_token,
    signing_secret=slack_signing_secret,
    process_before_response=True,
)


# Slackイベントハンドラー：Slackアプリがメンションされた時
@app.event("app_mention")
def handle_app_mention_events(event, say):
    result = say("少々お待ちください...")

    channel_id = event["channel"]
    input_text = re.sub("<@.+>", "", event["text"]).strip()

    sqs.send_message(
        QueueUrl=sqs_queue_url,
        MessageBody=json.dumps({
            "channel_id": channel_id,
            "input_text": input_text,
        }),
    )


# Lambdaイベントハンドラー
def lambda_handler(event, context):
    slack_handler = SlackRequestHandler(app=app)
    return slack_handler.handle(event, context)

Lambda関数の更新 (Bedrockを呼び出すバックエンド処理)

次に、「Bedrockを呼び出してSlackへ応答を返す」処理を行うLambda関数です。

こちらは、大幅に変更していますので、この後、詳しく解説します。

from slack_sdk import WebClient
import base64
import boto3
import json
import os


slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")

bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")

client = WebClient(token=slack_bot_token)


# Bedrockを使ってテキストから英語に翻訳する
def translate_to_english(input_text):
    # メッセージをセット
    messages = [
        {
            "role": "user",
            "content": f"与えられたテキストを英語に翻訳してください。翻訳されたテキストのみを出力してください。\n\n<text>{input_text}</text>",
        },
    ]

    # リクエストBODYをセット
    request_body = json.dumps({
        "messages": messages,
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
    })

    # Bedrock APIを呼び出す
    response = bedrock_runtime.invoke_model(
        modelId="anthropic.claude-3-haiku-20240307-v1:0",
        accept="application/json",
        contentType="application/json",
        body=request_body,
    )

    # レスポンスBODYから応答テキストを取り出す
    response_body = json.loads(response.get("body").read())
    output_text = response_body.get("content")[0].get("text")

    # 応答テキストを戻り値として返す
    return output_text


# Bedrockを使って画像を生成する
def generate_image(input_text):
    # リクエストBODYをセット
    request_body = json.dumps({
        "taskType": "TEXT_IMAGE",
        "textToImageParams": {
            "text": input_text,
        },
        "imageGenerationConfig": {
            "numberOfImages": 1,
            "height": 512,
            "width": 512,
            "quality": "standard",
            "cfgScale": 8,
            "seed": 0,
        },
    })

    # Bedrock APIを呼び出す
    response = bedrock_runtime.invoke_model(
        modelId="amazon.titan-image-generator-v1",
        accept="application/json",
        contentType="application/json",
        body=request_body,
    )

    # レスポンスBODYから画像データを取り出す
    response_body = json.loads(response.get("body").read())
    image_base64 = response_body.get("images")[0]
    image_data = base64.b64decode(image_base64.encode("ascii"))

    # 画像データを戻り値として返す
    return image_data


# Lambdaイベントハンドラー
def lambda_handler(event, context):
    # SQSキューから情報を取り出す
    body = json.loads(event["Records"][0]["body"])
    channel_id = body.get("channel_id")
    input_text = body.get("input_text")

    # 入力テキストを英語に翻訳する
    input_text_english = translate_to_english(input_text)

    # 入力テキストの英語への翻訳結果をSlackへ書き込む (デバッグ用)
    result = client.chat_postMessage(
        channel=channel_id,
        text=f"ご依頼内容は英語で次の通りです:\n{input_text_english}",
    )

    # Bedrockを呼び出して入力テキスト対する画像を生成する
    image_data = generate_image(input_text_english)

    # Slackへ画像をアップロードする
    result = client.files_upload_v2(
        channel=channel_id,
        content=image_data,
        filename="image.png",
        initial_comment="ご依頼の画像を作成しました。",
    )

入力されたテキストから画像を生成してSlackへ返すには、以下の順番に処理を行います。

• SQSキューから情報を取り出す
• Bedrockの「Titan Image Generator」モデルを使って、入力テキストに対する画像を生成する (generate_image)
• 生成された画像をBase64デコードして、Slackのチャンネルにアップロードする

しかし、「Titan Image Generator」は日本語の指示に対する精度がイマイチだと感じましたので、日本語の入力テキストを英語に翻訳してから「Titan Image Generator」に渡すようにしました。

• SQSキューから情報を取り出す
• Bedrockの「Claude 3 Haiku」モデルを使って、入力テキストを英語に翻訳する(translate_to_english)
• Bedrockの「Titan Image Generator」モデルを使って、入力テキストに対する画像を生成する (generate_image)
• 生成された画像をBase64デコードして、Slackのチャンネルにアップロードする

各処理のポイントを解説します。

Bedrockを使った画像生成

今回、画像を生成するためにAmazon提供の「Titan Image Generator」モデルを使用しています。

「Titan Image Generator」を使って画像を生成する際は、以下の内容をリクエストボディに与える必要があります。

{
    "taskType": "＜タスクの種類＞",
    "textToImageParams": {
        "text": "＜生成する画像を指示するプロンプトテキスト＞",
        "negativeText": "＜画像に含めて欲しくない物を指示するプロンプトテキスト＞"
    },
    "imageGenerationConfig": {
        "numberOfImages": ＜生成する画像の点数＞,
        "height": ＜画像の縦サイズ (ピクセル単位)＞,
        "width": ＜画像の横サイズ (ピクセル単位)＞,
        "quality": "＜画像の品質 (standard|premium)＞",
        "cfgScale": ＜プロンプトへの忠実さ (値が小さいほどランダム性が強くなる)＞,
        "seed": ＜画像生成の制御と再現に使用するシード値＞
    }
}

taskTypeには、「プロンプトで指示された画像を生成する」タスクであるTEXT_IMAGEを指定します。 この他にも「与えた画像の一部を変更する」といったタスクも用意されています。

textToImageParamsはタスクタイプにTEXT_IMAGEを指定した時に必要なパラメーター群で、text(画像生成の指示内容) とnegativeText(画像に含めて欲しくない物の指示) が指定できます。 今回はtextのみを使用しています。

imageGenerationConfigには生成する画像の属性を指定します。 パラメーターの詳細はAWS公式ドキュメントを参照してください。

Amazon Titan Image Generator G1 - Amazon Bedrock
https://docs.aws.amazon.com/ja_jp/bedrock/latest/userguide/model-parameters-titan-image.html

「Titan Image Generator」に対するリクエストのレスポンスは以下のようになります：

{
    "ResponseMetadata": {
        (省略)
    },
    "contentType": "application/json",
    "body": "＜レスポンスBody＞"
}

レスポンスBodyの内容は以下のようになります：

{
    "images": [
        "＜Base64でエンコードされた画像データ＞"
    ],
    "error": "＜エラーが発生した時のメッセージ (エラーが無ければnull)＞"
}

これらを踏まえて、得られたレスポンスから画像のバイナリデータを取り出します。

Slackへの画像のアップロード

Slackアプリを使って画像をチャンネルに投稿させるには、「ファイルのアップロード」を行う必要があります。

ファイルのアップロードは、かつては「files.uploadAPI」を使っていましたが、現在は「非推奨」となっており、代わりに「files.getUploadURLExternalAPI」と「files.completeUploadExternalAPI」を使うことになっています。

ただし、Slack SDKでは「files_uploadメソッド」(非推奨) に使い勝手が近い「files_upload_v2メソッド」が提供されているため、こちらを使います。

Slackへのファイルのアップロードは、以下のようなコードで行います。

    # Slackへ画像をアップロードする
    result = client.files_upload_v2(
        channel=channel_id,
        content=image_data,
        filename="image.png",
        initial_comment="ご依頼の画像を作成しました。",
    )

チャンネルの指定について、古いメソッド (files_upload) ではchannelsとなっており複数のチャンネルを同時に指定することができましたが、v2メソッドではchannelに変更されています。

アップロードするファイルのデータを指定する方法はfile(ローカルファイルのパスを指定) とcontent(ファイルのバイナリデータを直接指定) の2通りありますが、今回は後者を使用しています。

ファイル名について、「Titan Image Generator」を使った画像生成では「png」形式のデータが生成されるようでしたので、「image.png」というファイル名にしています。 (真面目にやるのであれば、生成されたファイルの形式を判定するロジックが必要になるかもしれません)

initial_commentパラメーターで、ファイルのアップロードと同時に書き込むメッセージを指定できます。

これで「画像ファイルを生成してくれるチャットボット」が完成しました。

おわりに

Amazon BedrockとSlackを使った生成AIチャットボットで、テキストのみのやり取りだけではなく「画像の取り扱い」を実践することができました。

パラメーターの設定やプロンプトの工夫によって、いろいろな可能性が考えられますので、是非みなさんも試してみてください。